package io.confluent.examples.streams

import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord, ConsumerRecords, KafkaConsumer}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.{IntegerDeserializer, IntegerSerializer}

import scala.collection.JavaConverters._
import scala.collection.mutable.ListBuffer


/**
  * This is a sample driver for the {@link SumLambdaExample}.
  * To run this driver please first refer to the instructions in {@link SumLambdaExample}.
  * You can then run this class directly in your IDE or via the command line.
  * <p>
  * To run via the command line you might want to package as a fatjar first. Please refer to:
  * <a href='https://github.com/confluentinc/kafka-streams-examples#packaging-and-running'>Packaging</a>
  * <p>
  * Once packaged you can then run:
  * <pre>
  * {@code
  * $ java -cp target/kafka-streams-examples-3.3.0-standalone.jar io.confluent.examples.streams.SumLambdaExampleDriver
 * }</pre>
  * You should terminate with {@code Ctrl-C}.
  * Please refer to {@link SumLambdaExample} for instructions on running the example.
  * <p>
  * Note: This example uses lambda expressions and thus works with Java 8+ only.
  */
object SumLambdaExampleDriver {
  def main(args: Array[String]): Unit = {
    val bootstrapServers = "10.100.189.30:9092"
    produceInput(bootstrapServers)
    consumeOutput(bootstrapServers)
  }

  def consumeOutput(bootstrapServers: String): Unit = {
    val properties: Properties = new Properties()
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[IntegerDeserializer])
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[IntegerDeserializer])
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "sum-lambda-example-consumer")
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

    val consumer: KafkaConsumer[Integer, Integer] = new KafkaConsumer[Integer, Integer](properties)
    consumer.subscribe(List(SumLambdaExample.SUM_OF_ODD_NUMBERS_TOPIC).asJava)
    while (true) {
      val records: ConsumerRecords[Integer, Integer] = consumer.poll(Long.MaxValue)
      for (record: ConsumerRecord[Integer, Integer] <- records.iterator().asScala) {
        println(s"Current sum of odd numbers is ${record.value()}")
      }
    }
  }

  def produceInput(bootstrapServers: String): Unit = {
    val properties: Properties = new Properties()
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[IntegerSerializer])
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[IntegerSerializer])
    val producer: KafkaProducer[Integer, Integer] = new KafkaProducer[Integer, Integer](properties)
    (0 to 100)
      .map(value => new ProducerRecord[Integer, Integer](SumLambdaExample.NUMBERS_TOPIC, value, value))
      .foreach(producer send _)

    producer.flush()


  }
}
